package com.bw.gmall.realtime.app.dwd.log2;

import com.bw.gmall.realtime.utils.MyKafkaUtil;
import com.bw.gmall.realtime.utils.MysqlUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class DwdTradeOrderRefund_a {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        tableEnv.executeSql(MyKafkaUtil.getTopicDb("topic_db_refund"));
       //读取退单表数据
        Table orderRefundInfo = tableEnv.sqlQuery(" select\n" +
                "data['id'] id,\n" +
                "data['user_id'] user_id,\n" +
                "data['order_id'] order_id,\n" +
                "data['sku_id'] sku_id,\n" +
                "data['refund_type'] refund_type,\n" +
                "data['refund_num'] refund_num,\n" +
                "data['refund_amount'] refund_amount,\n" +
                "data['refund_reason_type'] refund_reason_type,\n" +
                "data['refund_reason_txt'] refund_reason_txt,\n" +
                "data['create_time'] create_time,\n" +
                "pt \n" +
                "from topic_db\n" +
                "where `table` = 'order_refund_info'\n" +
                "and `type` = 'insert'\n"
        );
        tableEnv.createTemporaryView("order_refund_info",orderRefundInfo);
//        tableEnv.executeSql("select * from order_refund_info").print();

        //建立字典表
        tableEnv.executeSql(MysqlUtil.getBaseDicLookUpDDL());

        //读取订单数据 条件筛选退单数据
        Table orderInfoRefund = tableEnv.sqlQuery("select\n" +
                "data['id'] id,\n" +
                "data['province_id'] province_id,\n" +
                "`old`\n" +
                "from topic_db\n" +
                "where `table` = 'order_info'\n" +
                "and `type` = 'update'\n" +
                "and data['order_status']='1005'\n" +
                "and `old`['order_status'] is not null"
        );
        tableEnv.createTemporaryView("order_info_refund", orderInfoRefund);

        //读取详情表
        //读取订单数据 条件筛选退单数据
        Table orderDetailRefund = tableEnv.sqlQuery("select\n" +
                "data['order_id'] order_id,\n" +
                "data['sku_id'] sku_id\n" +
                "from topic_db\n" +
                "where `table` = 'order_detail'\n"
                + "and `type` = 'insert'\n"
        );
        tableEnv.createTemporaryView("order_detail", orderDetailRefund);

        Table orderShopRefund = tableEnv.sqlQuery("select\n" +
                        "data['sku_id'] sku_id,\n" +
                        "data['shop_id'] shop_id\n" +
                        "from topic_db\n" +
                        "where `table` = 'shop_sku'\n"
                + "and `type` = 'insert'\n"
        );
        tableEnv.createTemporaryView("shop_sku", orderShopRefund);



// 6. 构建宽表（关联所有表，新增 order_detail 和 shop_sku）
        Table resultTable = tableEnv.sqlQuery("select \n" +
                "ri.id,\n" +
                "ri.user_id,\n" +
                "ri.order_id,\n" +
                "ri.sku_id,\n" +
                "oi.province_id,\n" +
                "ss.shop_id, \n" +
                "date_format(ri.create_time, 'yyyy-MM-dd') date_id,\n" +
                "ri.create_time,\n" +
                "ri.refund_type as refund_type_code,\n" +  // 明确字段别名
                "type_dic.dic_name as refund_type_name,\n" +
                "ri.refund_reason_type as refund_reason_type_code,\n" +
                "reason_dic.dic_name as refund_reason_type_name,\n" +
                "ri.refund_reason_txt,\n" +
                "ri.refund_num,\n" +
                "ri.refund_amount,\n" +
                "ri.pt,\n" +
                "current_row_timestamp() row_op_ts\n" +
                "from order_refund_info ri\n" +
                // 关联订单表（获取省份ID）
                "join order_info_refund oi\n" +
                "on ri.order_id = oi.id\n" +
                // 关联订单详情表（验证订单与商品关联）
                "join order_detail od\n" +
                "on ri.order_id = od.order_id and ri.sku_id = od.sku_id\n" +  // 双重关联，确保准确性
                // 关联商品-店铺表（获取店铺ID）
                "join shop_sku ss\n" +
                "on ri.sku_id = ss.sku_id\n" +  // 通过 sku_id 关联店铺
                // 关联字典表（退款类型）
                "join base_dic for system_time as of ri.pt as type_dic\n" +
                "on ri.refund_type = type_dic.dic_code\n" +
                // 关联字典表（退款原因）
                "join base_dic for system_time as of ri.pt as reason_dic\n" +
                "on ri.refund_reason_type = reason_dic.dic_code"
        );
        tableEnv.createTemporaryView("result_table", resultTable);

        // 7. 创建 DWD 层输出表（新增 shop_id 字段）
        tableEnv.executeSql("create table dwd_trade_order_refund(\n" +
                "id string,\n" +
                "user_id string,\n" +
                "order_id string,\n" +
                "sku_id string,\n" +
                "province_id string,\n" +
                "shop_id string, \n" +
                "date_id string,\n" +
                "create_time string,\n" +
                "refund_type_code string,\n" +
                "refund_type_name string,\n" +
                "refund_reason_type_code string,\n" +
                "refund_reason_type_name string,\n" +
                "refund_reason_txt string,\n" +
                "refund_num string,\n" +
                "refund_amount string,\n" +
                "pt timestamp_ltz(3) ,\n" +
                "row_op_ts timestamp_ltz(3)\n" +
                ")" + MyKafkaUtil.getKafkaSinkDDL("dwd_trade_order_refund_a"));

        tableEnv.executeSql(" " +
                "insert into dwd_trade_order_refund select * from result_table");

        tableEnv.executeSql("select * from dwd_trade_order_refund").print();

        //获取宽表
//        Table resultTable = tableEnv.sqlQuery("select \n" +
//                "ri.id,\n" +
//                "ri.user_id,\n" +
//                "ri.order_id,\n" +
//                "ri.sku_id,\n" +
//                "oi.province_id,\n" +
//                "date_format(ri.create_time,'yyyy-MM-dd') date_id,\n" +
//                "ri.create_time,\n" +
//                "ri.refund_type,\n" +
//                "type_dic.dic_name,\n" +
//                "ri.refund_reason_type,\n" +
//                "reason_dic.dic_name,\n" +
//                "ri.refund_reason_txt,\n" +
//                "ri.refund_num,\n" +
//                "ri.refund_amount,\n" +
//                "ri.ts,\n" +
//                "current_row_timestamp() row_op_ts\n" +
//                "from order_refund_info ri\n" +
//                "join \n" +
//                "order_info_refund oi\n" +
//                "on ri.order_id = oi.id\n" +
//                "join \n" +
//                // 处理ri.pt时间戳
//                "base_dic for system_time as of ri.pt as type_dic\n" +
//                "on ri.refund_type = type_dic.dic_code\n" +
//                "join\n" +
//                "base_dic for system_time as of ri.pt as reason_dic\n" +
//                "on ri.refund_reason_type=reason_dic.dic_code"
//        );
//        tableEnv.createTemporaryView("result_table", resultTable);
//        tableEnv.executeSql("create table dwd_trade_order_refund(\n" +
//                "id string,\n" +
//                "user_id string,\n" +
//                "order_id string,\n" +
//                "sku_id string,\n" +
//                "province_id string,\n" +
//                "date_id string,\n" +
//                "create_time string,\n" +
//                "refund_type_code string,\n" +
//                "refund_type_name string,\n" +
//                "refund_reason_type_code string,\n" +
//                "refund_reason_type_name string,\n" +
//                "refund_reason_txt string,\n" +
//                "refund_num string,\n" +
//                "refund_amount string,\n" +
//                "ts string,\n" +
//                "row_op_ts timestamp_ltz(3)\n" +
//                ")" + MyKafkaUtil.getKafkaSinkDDL("dwd_trade_order_refund"));
//
//        tableEnv.executeSql(" " +
//                "insert into dwd_trade_order_refund select * from result_table");


//        env.execute();
    }
}
